package com.intct.dws;

import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.LocusRedisBean;
import com.intct.func.DwsHBaseSinkFunc;
import com.intct.func.RedisSinkFunc;
import com.intct.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gufg
 * @since 2025-07-04 14:06
 * 处理车辆轨迹数据
 * {
 *   "driverId": "11",
 *   "driverName": "邵5bfb8fd1c7af69efee217c72fabec38b",
 *   "lat": "30.70332",
 *   "lng": "104.05368",
 *   "orderId": "17592719043682",
 *   "orderIdDesensitization": "175********682",
 *   "time": "1475976805"
 * }
 *
 * 存储到Redis ==> 前端 ==> 选择数据类型 (set和zset)
 *      zset  1）那个字段 做为分数？？？ 字段选择time字段  2）值(经、纬度) lng+","+lat 3) key 订单
 *      set   1) key (当--curr、历--his) 2) 订单号
 *      订单结束："end" time字段=9999999999   经、纬度
 */
public class DwsLocusApp {
    public static void main(String[] args) throws Exception {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        // 启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 读取DWD(kafka)数据
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource("dwd_locus", "dwd_locus_group_id");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dws_kafka_source");

        // 处理数据
        SingleOutputStreamOperator<LocusRedisBean> mapDS = sourceDS.map(new MapFunction<String, LocusRedisBean>() {
            @Override
            public LocusRedisBean map(String jsonStr) throws Exception {
                /*
                 * {
                 *   "driverId": "11",
                 *   "driverName": "邵5bfb8fd1c7af69efee217c72fabec38b",
                 *   "lat": "30.70332",
                 *   "lng": "104.05368",
                 *   "orderId": "17592719043682",
                 *   "orderIdDesensitization": "175********682",
                 *   "time": "1475976805"
                 * }
                 */
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);

                Long score = 9999999999L;
                boolean isGameOver = true;
                // 判断订单是否结束
                if (!"end".equals(jsonObj.getString("driverId"))) {
                    score = Long.valueOf(jsonObj.getString("time"));
                    isGameOver = false;
                }

                LocusRedisBean locusRedisBean = LocusRedisBean.builder()
                        .orderId(jsonObj.getString("orderId"))
                        .score(score)
                        .value(jsonObj.getString("lng") + "," + jsonObj.getString("lat"))
                        .isGameOver(isGameOver)

                        .build();

                return locusRedisBean;
            }
        });

        // 输出到Redis
        mapDS.addSink(new RedisSinkFunc());

        // 输出到HBase
        sourceDS.addSink(new DwsHBaseSinkFunc());


        // 启动作业
        env.execute("dws_locus_job");
    }
}
